// Copyright (c) 2012, the Riak-Dart project authors (see AUTHORS file).
// All rights reserved. Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

/** HTTP client implementation */
class HttpClient implements Client {

  static final String RIAK_VCLOCK = "X-Riak-Vclock";
  static final String META_PREFIX = "X-Riak-Meta-";
  static final String INDEX_PREFIX = "X-Riak-Index-";

  String _host;
  int _port;
  io.HttpClient _client;

  HttpClient(this._host, this._port) {
    _client = new io.HttpClient();
  }

  Future<Fetched> fetch(ID target, [bool stream = false]) {
    Completer<Fetched> completer = new Completer();

    io.HttpClientConnection conn = _client.open("GET", _host, _port,
      "/buckets/${ uri.encodeUriComponent(target.bucket) }/keys/${ uri.encodeUriComponent(target.key) }");

    // TODO: handle etag requests
    // TODO: handle not-modified-since requests
    conn.onResponse = (io.HttpClientResponse clientResponse) {
      Fetched resp = new Fetched();
      resp.statusCode = clientResponse.statusCode;
      resp.success = false;
      resp.header = new ID(target.bucket, target.key);

      if (resp.statusCode == 200 || resp.statusCode == 304) {
        resp.success = true;
        resp.header.etag = clientResponse.headers.value(io.HttpHeaders.ETAG);
        resp.header.vclock = clientResponse.headers.value(RIAK_VCLOCK);

        String contentType = clientResponse.headers.value(io.HttpHeaders.CONTENT_TYPE);
        resp.content = new Content.stream(contentType, clientResponse.inputStream);
      } else if (resp.statusCode == 404) {
        resp.header.notModifiedSince(1); // we set this to not-modified for an hour only
      }

      if (!stream && resp.success) {
        String mediaType = resp.content.mediaType;
        bool isString = mediaType == Content.JSON || mediaType.startsWith("text/");

        io.InputStream input = resp.content.value;

        Future f = isString ? StreamUtils.readString(input) : StreamUtils.readString(input);
        f.handleException((ex) {
          if (!completer.future.isComplete) {
            completer.completeException(ex);
            return true;
          }
        });
        f.then((var o) { resp.content.value = o; completer.complete(resp); });
      } else {
        completer.complete(resp);
      }
    };
    conn.onError = (e) {
      if (completer.future.isComplete) {
        throw e;
      } else {
        completer.completeException(e);
      }
    };

    return completer.future;
  }

  void _setClientRequestHeaders(io.HttpClientRequest clientRequest, ID target) {
    bool filtered = false;
    if (target.vclock != null) {
      clientRequest.headers.set(RIAK_VCLOCK, target.vclock);
      filtered = true;
    }
    if (target.etag != null) {
      clientRequest.headers.set(io.HttpHeaders.IF_MATCH, target.etag);
      filtered = true;
    }
    if (!filtered && target.modified != null) {
      clientRequest.headers.set(io.HttpHeaders.IF_UNMODIFIED_SINCE, target.modified.toString());
    }
  }

  Future<Stored> store(ID target, Content content) {
    Completer<Stored> completer = new Completer();

    io.HttpClientConnection conn = _client.open("POST", _host, _port,
      "/buckets/${ uri.encodeUriComponent(target.bucket) }/keys/${ uri.encodeUriComponent(target.key) }");

    conn.onRequest = (io.HttpClientRequest clientRequest) {
      _setClientRequestHeaders(clientRequest, target);

      // content headers
      clientRequest.headers.set(io.HttpHeaders.CONTENT_TYPE, content.mediaType);
      if (content.hasIndex()) {
        for (String key in content.index.getKeys()) {
          List<Object> values = content.index.getIndexValues(key);
          String value = Strings.join(values.map((Object o) { return o.toString(); }), ", ");
          clientRequest.headers.set("${INDEX_PREFIX}${key}", value);
        }
      }

      io.OutputStream out = clientRequest.outputStream;
      out.onError = (e) {
        if (!completer.future.isComplete) {
          completer.completeException(e);
        } else {
          throw e;
        }
      };

      if (content.value is String) {
        out.writeString(content.value);
        out.close();
      } else if (content.value is io.InputStream) {
        io.InputStream stream = content.value;
        stream.pipe(out);
      } else {
        throw new Exception("Content value is not a String nor an InputStream");
      }
    };
    conn.onResponse = (io.HttpClientResponse clientResponse) {
      Stored resp = new Stored();
      resp.responseCode = clientResponse.statusCode;
      resp.success = false;

      if (resp.responseCode == 200 || resp.responseCode == 201 || resp.responseCode == 204) {
        resp.success = true;
      }
      completer.complete(resp);
    };
    conn.onError = (e) {
      if (!completer.future.isComplete) {
        completer.completeException(e);
      } else {
        throw e;
      }
    };

    return completer.future;
  }

  /** Delete the specified riak object */
  Future<Deleted> delete(ID target) {
    Completer<Deleted> completer = new Completer();

    io.HttpClient client = new io.HttpClient();
    io.HttpClientConnection conn = client.open("DELETE", _host, _port,
      "/buckets/${ uri.encodeUriComponent(target.bucket) }/keys/${ uri.encodeUriComponent(target.key) }");

    conn.onRequest = (io.HttpClientRequest clientRequest) {
      _setClientRequestHeaders(clientRequest, target);
    };
    conn.onResponse = (io.HttpClientResponse clientResponse) {
      Deleted resp = new Deleted();
      resp.responseCode = clientResponse.statusCode;
      resp.success = false;

      if (resp.responseCode == 200 || resp.responseCode == 204 || resp.responseCode == 404) {
        resp.success = true;
      }
      completer.complete(resp);
    };
    conn.onError = (e) {
      if (!completer.future.isComplete) {
        completer.completeException(e);
      } else {
        throw e;
      }
    };

    return completer.future;
  }

  Future<Queried> indexQuery(IndexQuery query, [Future<Dynamic> callback(String key)]) {
    Completer<Queried> completer = new Completer();

    String condition = query.end == null
        ? uri.encodeUriComponent(query.start.toString())
        : "${uri.encodeUriComponent(query.start.toString())}/${uri.encodeUriComponent(query.end.toString())}";

    io.HttpClient client = new io.HttpClient();
    io.HttpClientConnection conn = client.open("GET", _host, _port,
      "/buckets/${ uri.encodeUriComponent(query.bucket) }/index/${ uri.encodeUriComponent(query.index) }/$condition");

    conn.onResponse = (io.HttpClientResponse clientResponse) {
      Queried resp = new Queried();
      resp.responseCode = clientResponse.statusCode;
      resp.success = false;

      if (resp.responseCode == 200) {
        resp.success = true;
      } else {
        completer.complete(resp);
        return;
      }

      if (callback == null) {
        resp.keys = [];
        callback = (String key) { resp.keys.add(key); return null; };
      }

      // TODO: write streaming processing
      Future<String> f = StreamUtils.readString(clientResponse.inputStream);
      f.handleException((ex) {
        if (!completer.future.isComplete) {
          completer.completeException(ex);
          return true;
        }
      });
      f.then((String httpResultContent) {
        List futures = new List();

        Map m = json.JSON.parse(httpResultContent);
        List keysList = m == null ? null : m['keys'];
        resp.counter = 0;
        if (keysList != null) {
          for (String key in keysList) {
            resp.counter ++;
            Future callbackFuture = callback(key);
            if (callbackFuture != null)
              futures.add(callbackFuture);
          }
        }

        if (futures.length == 0) {
          completer.complete(resp);
        } else {
          Future flist = Futures.wait(futures);
          flist.handleException((e) {
            if (!completer.future.isComplete) {
              completer.completeException(e);
            } else {
              throw e;
            }
          });
          flist.then((results) { completer.complete(resp); });
        }
      });
    };
    conn.onError = (e) {
      if (!completer.future.isComplete) {
        completer.completeException(e);
      } else {
        throw e;
      }
    };

    return completer.future;
  }

  void close() {
    _client.shutdown();
  }
}

class StreamUtils {

  static Future<String> readString(io.InputStream input) {
    Completer<String> completer  = new Completer();
    StringBuffer textContent = new StringBuffer();

    io.StringInputStream stringInput = new io.StringInputStream(input);

    stringInput.onError = (e) {
      if (!completer.future.isComplete) {
        completer.completeException(e);
      } else {
        throw e;
      }
    };
    stringInput.onData = () { textContent.add(stringInput.read()); };
    stringInput.onClosed = () {
      String text = textContent.toString();
      completer.complete(text);
    };

    return completer.future;
  }

  static Future<List<int>> readBinary(io.InputStream input) {
    Completer<List<int>> completer = new Completer();
    List<int> result = [];

    input.onError = (e) {
      if (!completer.future.isComplete) {
        completer.completeException(e);
      } else {
        throw e;
      }
    };
    input.onData = () { result.addAll(input.read()); };
    input.onClosed = () { completer.complete(result); };

    return completer.future;
  }
}
